package hyl.core.struct;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.stream.Collectors;
import hyl.core.MyFun;
import hyl.core.run.IFilter;
import hyl.core.run.ISignalLamp;

/**
 * 该方法只适合少量的记录,记录多了以后影响性能<br>
 * 除非底层用c重写,否词快不起来的:)<br>
 * 未来的解决方法时把文件分块读取,分块处理,那就相当于重新做了一个hadoop内核,不值得,<br>
 * 我的定位是只做层的封装,底层先用别人的. 以后有机会时,自己重做底层,研究一下<br>
 * 
 * @author 阿友 3798955@qq.com
 *
 */
public class MyMapreduce {
	/**
	 * 获取几个键值的hashcode 因为hashcode 是动态计算的,所以这里是求当前的hashcode
	 * 
	 * @param map
	 * @param keys
	 * @return KeysHash 带有Map中维度字段值的 集合和 维度字段集合hashcode的对象
	 */
	@SuppressWarnings("rawtypes")
	public static KeysHash getKeysHash(Map map, String[] keys) {
		if (map == null || map.isEmpty())
			return null;
		if (MyFun.isEmpty(keys)) {
			KeysHash fhv = new KeysHash(0);
			return fhv;
		}
		final int len = keys.length;
		KeysHash fhv = new KeysHash(len);
		if (len > 0) {
			// 提取维度
			int[] 维度hc = new int[len];
			for (int i = 0; i < len; i++) {
				String s = keys[i];
				if (map.containsKey(s)) {
					Object obj = map.get(s);
					fhv._values[i] = obj;
					维度hc[i] = obj.hashCode();
				} else {
					fhv._values[i] = null;
					维度hc[i] = 0;
				}
			}
			fhv._code = MyFun.intarr2bytearr(维度hc); // getMapHashCode(obj1, 维度值);
		}
		return fhv;
	}

	/**
	 * 左联接
	 * 
	 * @param 表1
	 * @param 表2
	 * @param joinstr
	 * @return
	 */

	@SuppressWarnings("rawtypes")
	private static List<Map> join(List<Map> 表1, List<Map> 表2, String 连接串, boolean 是否左联接) {
		if (表1 == null && 表2 == null)
			return null;
		if (表1 == null && 表2 != null)
			return 表2;
		if (表1 != null && 表2 == null)
			return 表1;
		String[] joins = MyFun.split(连接串, ",");
		if (joins == null)
			return null;
		String[] key1 = new String[joins.length];
		String[] key2 = new String[joins.length];
		for (int k = 0; k < joins.length; k++) {
			String obj[] = MyFun.split(joins[k], "=");
			if (obj.length == 2) {
				key1[k] = obj[0];
				key2[k] = obj[1];
			}
		}
		List<Map> arr = new ArrayList<>();
		boolean flag = true;// 每个字段是否相同,假定相同
		boolean finded = false;// 没有找到匹配的记录
		for (Map item1 : 表1) {
			Map obj1 = (Map) item1;
			finded = false;
			for (Map item2 : 表2) {
				Map obj2 = (Map) item2;
				flag = true;
				for (int i = 0; i < key1.length; i++) {
					if (!MyCollect.eq(obj1.get(key1[i]), obj2.get(key2[i]))) {
						flag = false;// 有字段值不同,跳出
						break;
					}
				}
				if (flag) {// 所有字段值不同
					finded = true;// 找到了
					Map obj = merge(obj1, obj2);
					arr.add(obj);
					break;
				}
			}
			if (!finded && 是否左联接) { // 没有找到匹配的记录
				arr.add(obj1);
			}
		}
		return arr;
	}

	/*
	 * public static List<Map> innerjoin(List 表1, List 表2, String 连接串) { return
	 * join((List<Map>)表1,(List<Map>)表2, 连接串, false); } public static List<Map>
	 * leftjoin(List 表1, List 表2, String 连接串) { return
	 * join((List<Map>)表1,(List<Map>)表2, 连接串, true); }
	 */
	/**
	 * 内连接
	 * 
	 * @param 表1
	 * @param 表2
	 * @param 连接串
	 * @return
	 */
	@SuppressWarnings("rawtypes")
	public static List<Map> innerjoin(List<Map> 表1, List<Map> 表2, String 连接串) {
		return join(表1, 表2, 连接串, false);
	}

	/**
	 * 左联接
	 * 
	 * @param 表1
	 * @param 表2
	 * @param 连接串
	 * @return
	 */
	@SuppressWarnings("rawtypes")
	public static List<Map> leftjoin(List<Map> 表1, List<Map> 表2, String 连接串) {
		return join(表1, 表2, 连接串, true);
	}

	/**
	 * 根据属性,截取object对象的部分属性 生成新的对象 *
	 * 
	 * @param        {Object} object
	 * @param String cols 过滤字段集合，如果为空，那么不过滤
	 * @return {Object} 返回对象
	 */
	// 第一个是对象,第二个是筛选字段
	/*
	 * public Map section(Object... 集合) { if (集合.length == 2) return section(集合[0],
	 * 集合[1]); else if (集合.length == 1) return (Map) 集合[0]; else return null; }
	 */
	/**
	 * 选择字段
	 * 
	 * @param Map集合
	 * @param as别名  格式 "新名=原名,新名=原名,原名,原名..." ,注意=号两边不能有空格
	 * @return
	 */
	@SuppressWarnings({ "rawtypes" })
	public static Map select(Map 集合, String as别名) {
		if (MyFun.isEmpty(as别名))
			return 集合;
		return select(集合, as别名.split(","));
	}

	/**
	 * 选择字段
	 * 
	 * @param 集合
	 * @param as别名
	 * @return
	 */
	@SuppressWarnings({ "rawtypes", "unchecked" })
	public static Map select(Map 集合, String[] as别名) {
		Map n1 = new HashMap<>();
		if (集合 == null)
			return n1;
		if (as别名 == null || as别名.length == 0)
			return 集合;
		for (String name : as别名) {
			String[] arr = MyFun.split2(name, '=');
			// 如果分解两个项 =号左边是新字段别名,右边是原字段名
			if (arr != null) {
				n1.put(arr[0], 集合.get(arr[1]));
				continue;
			}
			n1.put(name, 集合.get(name));
		}
		return n1;
	}

	/**
	 * 选择字段
	 * 
	 * @param 集合
	 * @param as别名
	 * @return
	 */
	@SuppressWarnings("rawtypes")
	public static List<Map> select(List<Map> 集合, String as别名) {
		if (MyFun.isEmpty(as别名))
			return 集合;
		return select(集合, as别名.split(","));
	}

	/**
	 * 选择字段
	 * 
	 * @param 集合
	 * @param as别名
	 * @return
	 */
	@SuppressWarnings({ "rawtypes" })
	public static List<Map> select(List<Map> 集合, String[] as别名) {
		// selectMap
		List<Map> list = new ArrayList<>();
		for (Map mp : 集合) {
			list.add(select(mp, as别名));
		}
		return list;
	}

	/**
	 * 合并多个map 生成新的Map对象<br>
	 * 例如 : map2中含有和map1中相同的key，那么 map2中的值会覆盖掉map1中的值<br>
	 * 
	 * @param 待合并对象集合 ，
	 */
	@SuppressWarnings({ "unchecked", "rawtypes" })
	public static Map assign(Map... arguments) {
		Map n1 = new HashMap<>();
		if (arguments.length == 1)
			return arguments[0];
		else if (arguments.length < 1)
			return n1;
		else {
			for (int i = 0; i < arguments.length; i++) {
				if (arguments[i] == null)
					continue;
				n1.putAll(arguments[i]);
			}
		}
		return n1;
	}

	/**
	 * 合并的多个map <br>
	 * 有相同键时,后者不覆盖 前者,但是会新增一个键<br>
	 * 新增的键的名称会在右侧增加一个计数编号,表示新增的第几个键<br>
	 * 注意:如果有一样的键名会覆盖
	 * 
	 * @param arguments
	 * @return
	 * 
	 */
	@SuppressWarnings({ "unchecked", "rawtypes" })
	public static Map merge(Map... arguments) {
		Map n1 = new HashMap<>();
		if (arguments.length == 1)
			return arguments[0];
		else if (arguments.length < 1)
			return n1;
		else {
			for (int i = 0; i < arguments.length; i++) {
				if (arguments[i] == null)
					continue;
				Iterator<?> ite = arguments[i].entrySet().iterator();
				while (ite.hasNext()) {
					Map.Entry e = (Map.Entry) ite.next();
					if (n1.containsKey(e.getKey())) {
						// 如果两个值相同,可以忽略
						// 如果不同,新建一个别名存放数值
						if (!MyCollect.eq(n1.get(e.getKey()), e.getValue()))
							n1.put(e.getKey() + "_" + String.valueOf(i), e.getValue());// 更新别名
					} else
						n1.put(e.getKey(), e.getValue());
				}
			}
		}
		return n1;
	}

	/**
	 * 从表集合中 根据where 过滤函数返回结果，提取可用的记录，并返回
	 * 
	 * @param {Array} arr 数据集合
	 * @param {Object} where 过滤接口 如果return true 那么添加到结果集合
	 */
	@SuppressWarnings({ "rawtypes" })
	public static List<Map> where(List<Map> arr, IFilter<Map> filter) {
		if (arr == null || arr.isEmpty())
			return arr;
		if (filter == null)
			return arr;
		List<Map> arr1 = new ArrayList<>();// 目标数组
		for (Map mp : arr) {
			if (filter.where(mp))
				arr1.add(mp);
		}
		return arr1;
	}

	/**
	 * 私有函数 type==null||type==''||type=='a' 升序排列 否则 =='d'倒叙排列
	 * 
	 * @param {Object} prop
	 * @param {Object} type
	 */
	/**
	 * 根据order by 字段集合 排序table记录
	 * 
	 * @param {list<map>} table 格式[{},{}]
	 * @param {String} orderby 4种格式
	 *        分别表示升序,升序,升序,降序："field1%a,field3,field4%,field2%d"
	 */
	@SuppressWarnings({ "rawtypes" })
	public static List<Map> orderby(List<Map> table, String 排序字段) {
		if (MyFun.isEmpty(排序字段)) {
			return table;
		}
		return orderby(table, 排序字段.split(","));
	}

	/**
	 * 排序
	 * 
	 * @param table
	 * @param 排序字段
	 * @return
	 */
	@SuppressWarnings({ "rawtypes" })
	public static List<Map> orderby(List<Map> table, String[] 排序字段) {
		SortCompare sortcompare = new SortCompare();
		if (table == null)
			return null;
		if (table.isEmpty())
			return table;
		// String[] sp = orderby.split(",");
		for (int i = 排序字段.length - 1; i >= 0; i--) {
			String sp = 排序字段[i];
			int j = sp.indexOf('%');
			String field = null;
			String hz = null;
			if (j > 0) {
				field = sp.substring(0, j);
				hz = sp.substring(j);
			} else {
				field = sp.substring(0);
			}
			if (j > 0 && (hz.toUpperCase().endsWith("%D"))) {
				sortcompare.setConf(field, 'd');
				table.sort(sortcompare);// 降序排列
			} else {
				sortcompare.setConf(field, 'a');
				table.sort(sortcompare); // 升序排列 }
			}
		}
		return table;
	}

	/**
	 * 模仿group by count 功能 根据 分组字段 分组 <br>
	 * 从表记录集合 tarr中 统计 [统计表达式] 字段 <br>
	 * 包含 count计数，第一个值，最后一个值，最小，最大值，汇总sum值
	 * 
	 * @param {Array} tarr
	 * @param {String} groupcols group by 后面的字段
	 * @param {String} countcols 统计项中不能有 分组字段
	 *        "sum:field1","count:field2","min:field2"..，待统计的字段<br>
	 *        sum:,count:,max:,min:,first:,last:
	 */
	@SuppressWarnings({ "rawtypes" })
	public static List<Map> groupby(List<Map> table, String 统计表达式) {
		return groupby(table, null, 统计表达式, null);
	}

	/**
	 * 分组统计
	 * 
	 * @param table
	 * @param 分组字段
	 * @param 统计表达式
	 * @return
	 */
	@SuppressWarnings({ "rawtypes" })
	public static List<Map> groupby(List<Map> table, String 分组字段, String 统计表达式) {
		return groupby(table, 分组字段, 统计表达式, null);
	}

	/**
	 * 分组
	 * 
	 * @param table
	 * @param 分组字段
	 * @param 统计表达式
	 * @param 信号灯函数
	 * @return
	 */
	@SuppressWarnings({ "rawtypes" })
	public static List<Map> groupby(List<Map> table, String 分组字段, String 统计表达式, ISignalLamp<Map> 信号灯函数) {
		if (MyFun.isEmpty(统计表达式))
			return table;
		String[] g1 = null, g2 = null;
		if (!MyFun.isEmpty(分组字段)) {
			g1 = 分组字段.split(",");
		}
		if (!MyFun.isEmpty(统计表达式)) {
			g2 = 统计表达式.split(",");
		}
		return groupby(table, g1, g2, 信号灯函数);
	}

	@SuppressWarnings({ "rawtypes", "unchecked" })
	public static List<Map> groupby(List<Map> table, String[] 分组字段, String[] 统计表达式, ISignalLamp<Map> 信号灯函数) {
		if (table == null || table.isEmpty())
			return table;
		if (MyFun.isEmpty(统计表达式))
			return table;
		// List<Map> arr = orderby(table, 分组字段); //先排序
		TreeMap<byte[], Analayze> map1 = new TreeMap<>(new Comparator<byte[]>() {
			@Override
			public int compare(byte[] o1, byte[] o2) {
				return MyFun.compareBytes(o1, o2);
			}
		});
		Set<String> 分组字段ss = new HashSet<>();
		if (分组字段 != null) {
			for (String s : 分组字段) {
				if (MyFun.isEmpty(s))
					continue;
				分组字段ss.add(s);
			}
		}
		int 分组字段长度 = 分组字段ss.size();
		String[] 分组字段s = 分组字段ss.toArray(new String[分组字段长度]);
		Object[] 统计项s = new Object[统计表达式.length];
		int[] 表达类型 = new int[统计表达式.length];
		// List<Integer> 表达类型 = new ArrayList<>();
		// 解析 统计项表达式
		for (int i = 0; i < 统计表达式.length; i++) {
			String s = 统计表达式[i];
			// 0预留 表示禁止统计
			if (s.startsWith("count:")) {
				统计项s[i] = s.substring(6);
				表达类型[i] = 1;
			} else if (s.startsWith("sum:")) {
				统计项s[i] = s.substring(4);
				表达类型[i] = 2;
			} else if (s.startsWith("max:")) {
				统计项s[i] = s.substring(4);
				表达类型[i] = 3;
			} else if (s.startsWith("min:")) {
				统计项s[i] = s.substring(4);
				表达类型[i] = 4;
			} else if (s.startsWith("first:")) {
				统计项s[i] = s.substring(6);
				表达类型[i] = 5;
			} else if (s.startsWith("last:")) {
				统计项s[i] = s.substring(5);
				表达类型[i] = 6;
			} else {
				int j = s.indexOf(":");
				if (j < 0) {
					j = 0;
					统计项s[i] = s;
				} else
					统计项s[i] = s.substring(j + 1);
				表达类型[i] = 9;
			}
			// 不要avg: 可以用sum /count求
		}
		// 遍历原记录表 计算
		for (Map obj1 : table) {
			KeysHash fhv = getKeysHash(obj1, 分组字段);
			// 通过信号灯函数,我们可以过滤需要统计的项和不需要统计的项,
			// 间接实现了case when 功能 ,我是不是很厉害:) 哈哈哈
			if (null != 信号灯函数)
				表达类型 = 信号灯函数.run(obj1, 表达类型);
			// 如果存在分组对象,直接添加
			if (fhv == null)
				continue;
			if (map1.containsKey(fhv._code)) {
				Analayze anal = map1.get(fhv._code);
				Object[] cnt = anal._统计值集合;
				// 分项处理
				for (int i = 0; i < 统计项s.length; i++) {
					Object value = obj1.get(统计项s[i]);
					int 表达式 = 表达类型[i];
					switch (表达式) {
					case 0:
					case 5:
						break;
					case 1:
						// count
						int n = (int) cnt[i];
						cnt[i] = n + 1;
						break;
					case 2: // sum
						cnt[i] = MyFun.add(cnt[i], value);
						break;
					case 3: // max
						if (MyFun.compare(cnt[i], value) < 0)
							cnt[i] = value;
						break;
					case 4: // min
						if (MyFun.compare(cnt[i], value) > 0)
							cnt[i] = value;
						break;
					default: // last
						cnt[i] = value;
						break;
					}
				}
				// 底部计算 avg 函数可以通过 sum/count 实现
				/*
				 * for (int i = 0; i < 统计项s.length; i++) { Object value = obj1.get(统计项s[i]); int
				 * 表达式 = 表达类型[i]; switch (表达式) { case 6:// count Number[] cnt_c = (Number[])
				 * cnt[i]; int n = (int) cnt_c[0]; Number sum = (Number) cnt_c[1]; cnt[i] =
				 * (Number) MyFun.div(sum, n); break; } }
				 */
			} else// 如果没有该分组,添加分组对象(相当于初始化分组期初数值)
			{
				// 把维度值放进去,初始化统计项数组
				Analayze anal = new Analayze(fhv._values, 统计表达式.length);
				// 初始化统计项数值
				for (int i = 0; i < 统计项s.length; i++) {
					int 表达式 = 表达类型[i];
					switch (表达式) {
					case 0:
						break;
					case 1:// abg
						anal._统计值集合[i] = 1;
						break;
					default:
						Object val = obj1.get(统计项s[i]);
						anal._统计值集合[i] = val;
					}
					/*
					 * 原来是考虑 直接求avg ,后来放弃这种设计 case 6:// abg Number[] ar2 = new Number[2]; ar2[0] =
					 * 1; ar2[1] = (Number) val; anal._values[i] = ar2; break;
					 */
					/* else throw new Exception("缺少统计字段"); */
				}
				map1.put(fhv._code, anal);
			}
			// console.log( arr1.length)
		}
		List<Map> 结果集合 = new ArrayList<>();
		for (Map.Entry<byte[], Analayze> mp : map1.entrySet()) {
			Map map = new HashMap<>();
			Object[] x1 = mp.getValue()._维度值集合;
			Object[] x2 = mp.getValue()._统计值集合;
			for (int i = 0; i < 分组字段长度; i++) {
				map.put(分组字段s[i], x1[i]);
			}
			for (int i = 0; i < 统计表达式.length; i++) {
				map.put(统计表达式[i], x2[i]);
			}
			结果集合.add(map);
		}
		return 结果集合;
	}

	@SuppressWarnings({ "rawtypes" })
	public static List<Map> limit(List<Map> table, long from, long to) {
		return table.stream().skip(from).limit(to).collect(Collectors.toList());
	}

	/**
	 * 集合以指定字段去重后返回 注意非去重字段也会带出,你可以不管
	 */
	@SuppressWarnings({ "rawtypes" })
	public static List<Map> distinct(List<Map> table, String 去重字段) {
		String[] g1 = null;
		if (!MyFun.isEmpty(去重字段)) {
			g1 = 去重字段.split(",");
		}
		return distinct(table, g1);
	}

	@SuppressWarnings("rawtypes")
	public static List<Map> distinct(List<Map> table, String[] cols) {
		if (null == table)
			return null;
		if (MyFun.isEmpty(cols))
			return null;
		// Set<int[]> set1=new HashSet<>();
		TreeSet<byte[]> set1 = new TreeSet<>(new Comparator<byte[]>() {
			@Override
			public int compare(byte[] o1, byte[] o2) {
				return MyFun.compareBytes(o1, o2);
			}
		});
		List<Map> arr1 = new ArrayList<>();
		for (Map obj1 : table) {
			KeysHash fv = getKeysHash(obj1, cols);
			if (set1.contains(fv._code))
				continue;
			else {
				arr1.add(MyCollect.arr2Map(cols, fv._values));
				set1.add(fv._code);
			}
		}
		return arr1;
	}

	/**
	 * 
	 * @param table
	 * @param 去重字段
	 * @return 返回的结果集合会包含第一次匹配到的所有字段
	 */
	@SuppressWarnings({ "rawtypes" })
	public static List<Map> distinctExt(List<Map> table, String 去重字段) {
		String[] g1 = null;
		if (!MyFun.isEmpty(去重字段)) {
			g1 = 去重字段.split(",");
		}
		return distinct(table, g1);
	}

	/**
	 * 
	 * @param table
	 * @param 去重字段
	 * @return 返回的结果集合会包含第一次匹配到的所有字段
	 */
	@SuppressWarnings("rawtypes")
	public static List<Map> distinctExt(List<Map> table, String[] cols) {
		if (null == table)
			return null;
		if (MyFun.isEmpty(cols))
			return table;
		// Set<int[]> set1=new HashSet<>();
		TreeSet<byte[]> set1 = new TreeSet<>(new Comparator<byte[]>() {
			@Override
			public int compare(byte[] o1, byte[] o2) {
				return MyFun.compareBytes(o1, o2);
			}
		});
		List<Map> arr1 = new ArrayList<>();
		for (Map obj1 : table) {
			KeysHash fv = getKeysHash(obj1, cols);
			if (set1.contains(fv._code))
				continue;
			else {
				arr1.add(obj1);
				set1.add(fv._code);
			}
		}
		return arr1;
	}
	// group by ,where sum,count,max,min,avg,first,last,limit,distinct,sort
}

/**
 * 排序引擎
 * 
 * @author 37798955@qq.com
 *
 */
@SuppressWarnings({ "rawtypes" })
class SortCompare implements Comparator<Map> {
	String _prop;
	char _type;

	void setConf(String prop, char type) {
		_prop = prop;
		_type = type;
	}

	@Override
	public int compare(Map obj1, Map obj2) {
		Object val1 = obj1.get(_prop);
		Object val2 = obj2.get(_prop);
		int r = 0;
		if (MyFun.compare(val1, val2) < 0) {
			r = -1;
		} else if (MyFun.compare(val1, val2) > 0) {
			r = 1;
		}
		if (_type == 'd') {
			return 0 - r;
		} else
			return r;
	}
}

/**
 * 分析过程的临时对象
 * 
 * @author 阿友 3798955@qq.com
 *
 */
class Analayze {
	public Analayze(Object[] keys, int 统计项数) {
		_维度值集合 = keys;
		_统计值集合 = new Object[统计项数];
	}

	Object[] _维度值集合;
	Object[] _统计值集合;
}

/**
 * 保存hashcode 和 数组 ,配合getHashValue()函数使用
 * 
 * @author 阿友 3798955@qq.com
 *
 */
class KeysHash {
	public KeysHash(int len) {
		_values = new Object[len];
	}

	//
	byte[] _code = new byte[0];
	Object[] _values;
}